package com.atguigu.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;

public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        // Create runtime Environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);// 并行度和kafka主题的分区数相同

        // 设置检查点
//        env.setDefaultSavepointDirectory("hdfs://hadoop102:8020/gmall2022/");

        // 读取kafka数据 topic = topic_log
        String topic = "topic_log";
        String groupId = "topic_log_0927";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaSource(topic, groupId));

        // 读数据进行初步过滤
        OutputTag<String> dirtyOutput = new OutputTag<String>("dirty") {
        };
        SingleOutputStreamOperator<JSONObject> filterDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    ctx.output(dirtyOutput, value);
                }
            }
        });

        KeyedStream<JSONObject, String> keyedStream = filterDS.keyBy(value -> value.getJSONObject("common").getString("mid"));

        // 新老用户校验
        SingleOutputStreamOperator<JSONObject> jsonObjWithNewFlagDS = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {
            ValueState<String> state = null;
            SimpleDateFormat sdf;

            @Override
            public void open(Configuration parameters) throws Exception {
                state = getRuntimeContext().getState(new ValueStateDescriptor<String>("state", String.class));
                sdf = new SimpleDateFormat("yyyy-MM-dd");
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {

                JSONObject common = value.getJSONObject("common");

                if ("1".equals(common.getString("is_new"))) {
                    String is = state.value();
                    if (is != null) {
                        value.put("is_new", "0");
                    } else {
                        state.update("0");
                    }
                }

                return value;
            }
        });


        // 分流 主流page 测输出流error action display start
        OutputTag<String> startTag = new OutputTag<String>("start") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("error") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("action") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("display") {
        };

        SingleOutputStreamOperator<String> pageDS = jsonObjWithNewFlagDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {

                String jsonString = value.toJSONString();

                // 分流error数据
                JSONObject error = value.getJSONObject("err");
                if (error != null) {
                    ctx.output(errorTag, jsonString);
                }

                // 分流start的数据
                JSONObject start = value.getJSONObject("start");
                if (start != null) {
                    ctx.output(startTag, jsonString);
                } else {
                    // 不是start的数据都是主流page的数据
                    out.collect(jsonString);

                    Long ts = value.getLong("ts");
                    String page_id = value.getJSONObject("page").getString("page_id");
                    JSONObject common = value.getJSONObject("common");

                    // 筛选出display的数据
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null && displays.size() != 0) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            display.put("ts", ts);
                            display.put("page_id", page_id);
                            ctx.output(displayTag, display.toJSONString());
                        }
                    }

                    // 筛选出action的数据
                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null && actions.size() != 0) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            action.put("page_id", page_id);
                            action.put("common", common.toJSONString());

                            ctx.output(actionTag, action.toJSONString());
                        }
                    }
                }
            }
        });

        // 获取测输出流
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> errorDS = pageDS.getSideOutput(errorTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        DataStream<String> actionDS = pageDS.getSideOutput(actionTag);

        pageDS.print("Page>>>>");
        startDS.print("Start>>>>");
        errorDS.print("Error>>>>");
        displayDS.print("Display>>>>");
        actionDS.print("Action>>>>");

        // 存入kafka
        pageDS.addSink(MyKafkaUtil.getKafkaSink("dwd_page_log"));
        startDS.addSink(MyKafkaUtil.getKafkaSink("dwd_start_log"));
        errorDS.addSink(MyKafkaUtil.getKafkaSink("dwd_error_log"));
        displayDS.addSink(MyKafkaUtil.getKafkaSink("dwd_display_log"));
        actionDS.addSink(MyKafkaUtil.getKafkaSink("dwd_action_log"));

        // 启动任务
        env.execute();
    }
}
